package com.ld.security.common.rocketmq.config;

import com.ld.security.common.core.util.ReflectionUtil;
import com.ld.security.common.rocketmq.mq.SuperConsumer;
import com.ld.security.common.rocketmq.mq.SuperProducer;
import lombok.SneakyThrows;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.LinkedList;
import java.util.List;

/**
 *
 */
@Configuration
public class MqConfig {
    /**
     * mq 地址
     */
    @Value("${rocketmq.server-addr}")
    private String serverAddr;

    @Autowired(required = false)
    private List<SuperConsumer> superConsumerList;

    @Autowired(required = false)
    private List<SuperProducer> superProducers;

    @SneakyThrows
    @Bean
    public Object defaultMQPushConsumer(){
        if(ReflectionUtil.isNull(superConsumerList)) {
            return null;
        }
        LinkedList<DefaultMQPushConsumer> linkedList =  new LinkedList<>();

        for (SuperConsumer val : superConsumerList) {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(val.getProducerGroup());
            consumer.setMessageModel(MessageModel.BROADCASTING);
            consumer.setNamesrvAddr(serverAddr);
            consumer.subscribe(val.getTopic(), val.getSubExpression());

            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.registerMessageListener(val);
            consumer.start();
            linkedList.add(consumer);
        }
        return linkedList;

    }

    @SneakyThrows
    @Bean
    public Object defaultMQProducer(){
        if(ReflectionUtil.isNull(superProducers)){
            return null;
        }
        LinkedList<DefaultMQProducer> linkedList =  new LinkedList<>();

        for (SuperProducer val : superProducers) {
            DefaultMQProducer producer = new DefaultMQProducer(val.getProducerGroup());
            producer.setNamesrvAddr(serverAddr);
            val.setProducer(producer);
            producer.start();
            linkedList.add(producer);
        }
        return linkedList;
    }


}
